package com.triple.triplerpc.server.tcp;

import cn.hutool.core.util.IdUtil;
import com.triple.triplerpc.RpcApplication;
import com.triple.triplerpc.model.RpcRequest;
import com.triple.triplerpc.model.RpcResponse;
import com.triple.triplerpc.model.ServiceMetaInfo;
import com.triple.triplerpc.protocol.*;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetClient;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * Vertx TCP 请求客户端 - 消费者代理对象发送请求
 */
public class VertxTcpClient {

    public static RpcResponse doRequest(RpcRequest rpcRequest, ServiceMetaInfo serviceMetaInfo) throws InterruptedException, ExecutionException {
        // 发送 TCP 请求
        Vertx vertx = Vertx.vertx();
        NetClient netClient = vertx.createNetClient();
        // 使用 CompletableFuture 转异步为同步
        CompletableFuture<RpcResponse> responseFuture = new CompletableFuture<>();
        netClient.connect(serviceMetaInfo.getServicePort(), serviceMetaInfo.getServiceHost(),
                result -> {
                    if (result.succeeded()) {
                        System.out.println(" >>> 自定义协议 - 发送 TCP 请求...");
                        io.vertx.core.net.NetSocket socket = result.result();
                        // 构造消息
                        ProtocolMessage<RpcRequest> protocolMessage = new ProtocolMessage<>();
                        ProtocolMessage.Header header = new ProtocolMessage.Header();
                        header.setMagic(ProtocolConstant.PROTOCOL_MAGIC);
                        header.setVersion(ProtocolConstant.PROTOCOL_VERSION);
                        String serializerName = RpcApplication.getRpcConfig().getSerializer();
                        ProtocolMessageSerializerEnum protocolSerializer = ProtocolMessageSerializerEnum.getEnumByValue(serializerName);
                        if (protocolSerializer != null) {
                            header.setSerializer((byte) protocolSerializer.getKey());
                        } else {
                            // 自定义序列化器，用特殊值标记
                            header.setSerializer((byte) -1);
                        }
                        header.setType((byte) ProtocolMessageTypeEnum.REQUEST.getKey());
                        header.setRequestId(IdUtil.getSnowflakeNextId());
                        protocolMessage.setHeader(header);
                        protocolMessage.setBody(rpcRequest);
                        // 编码请求, 里面会序列化请求体
                        try {
                            Buffer encodeBuffer = ProtocolMessageEncoder.encode(protocolMessage);
                            socket.write(encodeBuffer);
                        } catch (IOException e) {
                            throw new RuntimeException("协议消息编码错误");
                        }
                        // 装饰者模式: 接收响应
                        TcpBufferHandlerWrapper bufferHandlerWrapper = new TcpBufferHandlerWrapper(buffer -> {
                            try {
                                ProtocolMessage<RpcResponse> rpcResponseProtocolMessage = (ProtocolMessage<RpcResponse>) ProtocolMessageDecoder.decode(buffer);
                                responseFuture.complete(rpcResponseProtocolMessage.getBody());
                            } catch (IOException e) {
                                throw new RuntimeException("协议消息解码错误");
                            }
                        });
                        socket.handler(bufferHandlerWrapper);
                    } else {
                        System.err.println("连接 TCP 服务器失败");
                        responseFuture.completeExceptionally(new RuntimeException("连接 TCP 服务器失败"));
                    }
                });
        // 阻塞, 直到响应完成, 才会继续向下执行
        RpcResponse rpcResponse = responseFuture.get();
        System.out.println(" >>> 收到响应, 解码后响应对象: " + rpcResponse.toString());
        // 记得关闭连接
        netClient.close();
        return rpcResponse;
    }

}
